package cn.dmp.tags

import cn.dmp.utils.{Constants, JedisPools}
import com.typesafe.config.ConfigFactory
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

import scala.collection.mutable.ListBuffer

/**
  * 将数据进行标签化处理
  * 作者：老杨
  */
object Tags4Context {

    def main(args: Array[String]): Unit = {

        val config = ConfigFactory.load()

        // 设置job所需的参数 sparkconf
        val sparkConf = new SparkConf()
        sparkConf.setAppName(s"${this.getClass.getSimpleName}")
        sparkConf.setMaster("local[*]")
        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        // sparkcontext
        val sc = new SparkContext(sparkConf)

        // 广播字典数据 -> appdict
        val appdictMap = sc.textFile(config.getString("appdict.path"))
          .map(_.split("\t", -1)).filter(_.length >= 5)
          .map(arr => (arr(4), arr(1))).collect().toMap

        val appdictBroadcast = sc.broadcast(appdictMap)

        // 停用词库
        val stopWords = sc.textFile(config.getString("stopwords.path")).map(stopword => (stopword, 0)).collect().toMap
        val stopwordsBroadcast = sc.broadcast(stopWords)


        val sQLContext = new SQLContext(sc)

        // 读取数据 -> parquet
        val dataFrame = sQLContext.read.parquet(config.getString("parquet.path"))

        dataFrame.where(Constants.haveUserIdLog).mapPartitions(itr => {

            val jedis = JedisPools.getJedis

            val list = ListBuffer[(String, List[(String, Int)])]()

            itr.foreach(row => {

                val adTags = Tags4Ad.makeTags(row)
                val appTags = Tags4App.makeTags(row, appdictBroadcast)
                val deviceTags = Tags4Device.makeTags(row)
                val kwTags = Tags4KeyWords.makeTags(row, stopwordsBroadcast)
                val areaTags = Tags4Area.makeTags(row)

                // 商圈标签
                val busTags = Tags4Business.makeTags(row, jedis)

                list.append((getUserId(row), (adTags ++ appTags ++ deviceTags ++ kwTags ++ areaTags++busTags).toList))
            })
            jedis.close()
            list.toIterator
        }).reduceByKey{
            case (list1, list2) => (list1 ++ list2).groupBy(_._1)
              //.mapValues(_.map(_._2).sum).toList
              .mapValues(_.foldLeft(0)(_ + _._2)).toList
        }.foreach(println)

        sc.stop()

    }


    def getUserId(row: Row): String = {
        row match {
            case v if !v.getAs[String]("imei").isEmpty => "IM:"+v.getAs[String]("imei")
            case v if !v.getAs[String]("idfa").isEmpty => "ID:"+v.getAs[String]("idfa")
            case v if !v.getAs[String]("mac").isEmpty => "MC:"+v.getAs[String]("mac")
            case v if !v.getAs[String]("androidid").isEmpty => "AID:"+v.getAs[String]("androidid")
            case v if !v.getAs[String]("openudid").isEmpty => "UID:"+v.getAs[String]("openudid")

            case v if !v.getAs[String]("imeimd5").isEmpty => "IMM:"+v.getAs[String]("imeimd5")
            case v if !v.getAs[String]("idfamd5").isEmpty => "IDM:"+v.getAs[String]("idfamd5")
            case v if !v.getAs[String]("macmd5").isEmpty => "MCM:"+v.getAs[String]("macmd5")
            case v if !v.getAs[String]("androididmd5").isEmpty => "AIDM:"+v.getAs[String]("androididmd5")
            case v if !v.getAs[String]("openudidmd5").isEmpty => "UIDM:"+v.getAs[String]("openudidmd5")

            case v if !v.getAs[String]("imeisha1").isEmpty => "IMS:"+v.getAs[String]("imeisha1")
            case v if !v.getAs[String]("idfasha1").isEmpty => "IDS:"+v.getAs[String]("idfasha1")
            case v if !v.getAs[String]("macsha1").isEmpty => "MCS:"+v.getAs[String]("macsha1")
            case v if !v.getAs[String]("androididsha1").isEmpty => "AIDS:"+v.getAs[String]("androididsha1")
            case v if !v.getAs[String]("openudidsha1").isEmpty => "UIDS:"+v.getAs[String]("openudidsha1")
        }
    }

}
